package com.atguigu.sparkstreaming.apps

import java.time.LocalDate
import java.util
import java.util.Properties

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.constants.{DBNameConstant, PrefixConstant, TopicConstant}
import com.atguigu.realtime.utils.{PropertiesUtil, RedisUtil}
import com.atguigu.sparkstreaming.beans.{ActionLog, CouponAlertInfo, OrderDetail, OrderInfo, ProvinceInfo, SaleDetail, UserInfo}
import com.atguigu.sparkstreaming.utils.{DStreamUtil, DateParseUtil}
import com.google.gson.Gson
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.util.control.Breaks

/**
 *
 *
 *   at least once + 幂等输出 保证精确一次。
 *
 *
 *   ------------------------------
 *
 *   购物明细:
 *            order_detail
 *                  left join  order_info  on order_detail.order_id = order_info.id
 *                    目的从 order_info获得当前这笔订单详情的province_id和user_id
 *
 *            使用province_id和user_id 到对应的维度表中去关联维度信息即可。
 *
 *
 *    原则：  事实表的关联，只能在流中操作。原因是事实表中的事实是源源不断地产生。
 *
 *          关联维度表时，区分所关联维度的类型:
 *             不用监控，在用的时候，直接去mysql查询即可,查询一次后，直接在App端保存:
 *                几乎不变(不会insert也不会update)的维度表:   时间，省份地区
 *
 *             实时监控变化:
 *                只会新增(sku_info):   当商家上架了新商品时，会向表中新增商品。但是商品的信息一旦入库，禁止修改！
 *                会变化的维度(既有insert还有update)：  user_info
 *
 *
 *---------------------------------------------
 *    ①在实时的流中消费 order_detail 和 order_info，进行关联
 *    ②监控实时insert和update的用户信息(维度表)，将信息从Mysql同步到redis
 *        关联用户信息时，只需要查询redis即可
 *    ③不变的维度，可以在App运行之前，一次性从mysql查询，缓存到App端。
 *          广播变量(分布式缓存)
 *          将从Driver端广播的数据，分发到每个Executor端进行缓存。在每个Executor运行的Task都可以从Executor端读取数据。
 *
 *
 *----------------------------------------------
 * Join的条件:
 *    ①两个流必须从同一个StreamingContext获取
 *    ②只有 DS[K,V]类型才能Join
 *      order_detail
 *                   left join  order_info  on order_detail.order_id = order_info.id
 *          把on中的字段作为 K
 *
 * ---------------------------------------
 * 无法Join的根本原因： 要Join的数据无法在同一个批次被消费到。
 *
 * 数量关系：  1个order_info 对应 N个order_detail
 *
 * 原则：  谁早到，谁就写入缓存。谁晚到，就去缓存找。
 *
 * 对于每一个批次的order_info
 *        ①和当前批次已经达到的order_detail关联
 *        ③到缓存中去之前已经早到的order_detail，找到就关联
 *        ②把自己写入缓存，以防后续有晚到的Order_detail
 *
 *  对于每一个批次的Order_detail
 *        ④和当前批次已经达到的order_info关联。如果Order_info做过此类操作，这里可以省略
 *        ⑤无法关联的order_detail需要读取缓存中早到的order_info，读到就关联
 *        ⑥如果 ⑤读不到，说明当前的order_detail早到了，将其写入缓存。等待后续批次到达的order_info，再关联
 *
 *
 *----------------------------------------
 * 缓存中数据的存储:
 * order_info:
 * 当前的需求是，对某个早到或晚到order_detail，根据order_id求订单信息
 *    key:  orderinfo:order_id
 *    value:  string
 *
 * Order_detail:
 *    key:  orderdetail:order_id
 *    value:  set
 *
 */
object SaleDetailApp extends BaseApp {
  override var groupId: String = "realtime220309"
  override var topic: String = TopicConstant.ORDER_DETAIL
  override var appName: String = "AlertApp"
  override var batchDuration: Int = 10
  var topic2: String = TopicConstant.ORDER_INFO

  /*
      查询Mysql

      JDBC:     获取连接，准备sql，预编译，查询结果得到resultSet，封装到Map中
      sparksql:   提供查询JDBC的方式，封装好
   */
  def queryProvinceInfo(sparkConf: SparkConf):mutable.Map[String,ProvinceInfo]={

    val provinceInfos: mutable.HashMap[String, ProvinceInfo] = new mutable.HashMap[String, ProvinceInfo]()

    // .getOrCreate()如果当前JVM已经有SparkContext就get，没有再create
    val sparkSession: SparkSession = SparkSession.builder.config(sparkConf).getOrCreate()

    val properties = new Properties

    properties.setProperty("user",PropertiesUtil.getProperty("jdbc.user"))
    properties.setProperty("password",PropertiesUtil.getProperty("jdbc.password"))
    properties.setProperty("driver",PropertiesUtil.getProperty("jdbc.driver.name"))

    //没类型    Dataset[Row]
    val dataFrame: DataFrame = sparkSession.read.jdbc(PropertiesUtil.getProperty("jdbc.url"), "base_province", properties)
    //转为有类型的 Dataset[ProvinceInfo]
    import sparkSession.implicits._

    val dataSet: Dataset[ProvinceInfo] = dataFrame.as[ProvinceInfo]

    dataSet.collect().foreach(provinceInfo => provinceInfos.put(provinceInfo.id ,provinceInfo))

    provinceInfos

  }


  def main(args: Array[String]): Unit = {

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName(appName)

    //加入ES的集群地址，端口，其他设置
    sparkConf.set("es.nodes",PropertiesUtil.getProperty("es.nodes"))
    sparkConf.set("es.port",PropertiesUtil.getProperty("es.port"))
    //运行自动创建index
    sparkConf.set("es.index.auto.create", "true")
    //允许将主机名转换为ip
    sparkConf.set("es.nodes.wan.only", "true")

    //重写context
    context = new StreamingContext(sparkConf,Seconds(batchDuration))

    //在Driver端查询省份维度
    val provinceMap: mutable.Map[String, ProvinceInfo] = queryProvinceInfo(sparkConf)

    //广播  广播后，不能用provinceMap，用provinceMapBC
    // 广播和闭包的区别:  广播是一个Executor复制一份  闭包是一个Task复制一份
    val provinceMapBC: Broadcast[mutable.Map[String, ProvinceInfo]] = context.sparkContext.broadcast(provinceMap)

    runApp{

      val orderDetailDS: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.createDStream(groupId, context, topic)
      val orderInfoDS: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.createDStream(groupId, context, topic2)

      //当前批次的偏移量
      var ODRanges: Array[OffsetRange] = null
      var OIRanges: Array[OffsetRange] = null

      val ds1: DStream[(String, OrderDetail)] = orderDetailDS.transform(rdd => {

        //获取当前消费到的这个批次偏移量
        ODRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //样例类转换
        rdd.map(record => {
          val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
          (orderDetail.order_id, orderDetail)
        })

      })

      val ds2: DStream[(String, OrderInfo)] = orderInfoDS.transform(rdd => {

        //获取当前消费到的这个批次偏移量
        OIRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //样例类转换
        rdd.map(record => {
          val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])
          orderInfo.create_date = DateParseUtil.parseDateTimeStrToDate(orderInfo.create_time)
          orderInfo.create_hour = DateParseUtil.parseDateTimeStrToHour(orderInfo.create_time)
          (orderInfo.id, orderInfo)
        })

      })

     //由于目前的网络延迟可能会造成 od 和 oi两个流中的数据无法正确匹配，要保证Join后依旧可以保留批次中的所有数据，再针对Join的情况进行处理
     val ds3: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = ds2.fullOuterJoin(ds1)

      val ds4: DStream[SaleDetail] = ds3.mapPartitions(partition => {

        val jedis: Jedis = RedisUtil.getJedis

        val gson = new Gson()

        //构造一个集合，放封装好的SaleDetail
        val result: ListBuffer[SaleDetail] = new ListBuffer[SaleDetail]()

        partition.foreach {
          case (orderId, (orderInfoOption, orderDetailOption)) => {

            if (orderInfoOption != None) {

              val orderInfo: OrderInfo = orderInfoOption.get

              //①和当前批次已经达到的order_detail关联
              if (orderDetailOption != None) {

                val orderDetail: OrderDetail = orderDetailOption.get

                result.append(new SaleDetail(orderInfo, orderDetail))
              }

              //②把自己写入缓存，以防后续有晚到的Order_detail
              jedis.setex(PrefixConstant.order_info_redis_preffix + orderId,PrefixConstant.max_delay_time, gson.toJson(orderInfo))

              // ③到缓存中去之前已经早到的order_detail，找到就关联
              // 如果redis中set类型对应的key不存在，返回一个 Set()，并不是null
              val orderDetailSet: util.Set[String] = jedis.smembers(PrefixConstant.order_detail_redis_preffix + orderId)

              orderDetailSet.forEach(orderDetailStr => {

                val od: OrderDetail = JSON.parseObject(orderDetailStr, classOf[OrderDetail])

                result.append(new SaleDetail(orderInfo, od))

              })

            } else {

              val orderDetail: OrderDetail = orderDetailOption.get

              //⑤无法关联的order_detail需要读取缓存中早到的order_info，读到就关联
              // string类型，如果key不存在，返回null
              val orderInfoStr: String = jedis.get(PrefixConstant.order_info_redis_preffix + orderId)

              if (orderInfoStr != null) {

                val orderInfo: OrderInfo = JSON.parseObject(orderInfoStr, classOf[OrderInfo])

                result.append(new SaleDetail(orderInfo, orderDetail))

              } else {
                //⑥如果 ⑤读不到，说明当前的order_detail早到了，将其写入缓存。等待后续批次到达的order_info，再关联
                jedis.sadd(PrefixConstant.order_detail_redis_preffix + orderId, gson.toJson(orderDetail))

                jedis.expire(PrefixConstant.order_detail_redis_preffix + orderId,PrefixConstant.max_delay_time)

              }

            }

          }
        }

        jedis.close()

        result.toIterator


      })

      //关联维度数据
      val ds5: DStream[SaleDetail] = ds4.mapPartitions(partition => {

        val jedis: Jedis = RedisUtil.getJedis

        val result: Iterator[SaleDetail] = partition.map(saleDetail => {

          val userInfoStr: String = jedis.get(PrefixConstant.user_info_redis_preffix + saleDetail.user_id)

          if (userInfoStr != null) {

            //合并用户信息
            saleDetail.mergeUserInfo(JSON.parseObject(userInfoStr, classOf[UserInfo]))

          }

          //关联省份
          saleDetail.mergeProvinceInfo(provinceMapBC.value.get(saleDetail.province_id).get)

          saleDetail

        })

        jedis.close()

        result


      })


      //写入ES
      //提供的那些静态方法导入
      import org.elasticsearch.spark._

      ds5.foreachRDD( rdd => {

        rdd.cache()

        println("即将写入:"+rdd.count())

        rdd.saveToEs(DBNameConstant.SALEDETAILINDEX + LocalDate.now(),Map( "es.mapping.id" -> "order_detail_id"))

        //初始DS提交偏移量
        orderDetailDS.asInstanceOf[CanCommitOffsets].commitAsync(ODRanges)
        orderInfoDS.asInstanceOf[CanCommitOffsets].commitAsync(OIRanges)


      })

    }


  }
}
